package com.boarsoft.message.sofamq;

import com.boarsoft.message.core.MessageService;
import com.boarsoft.message.sofamq.converter.SofaMsgConverter;
import io.openmessaging.api.Action;
import io.openmessaging.api.Message;
import io.openmessaging.api.ConsumeContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SofaMQConsumerEx extends SofaMQConsumer{

	private static final Logger log = LoggerFactory.getLogger(SofaMQConsumerEx.class);

	protected MessageService handler;

	protected SofaMsgConverter converter;

	@Override
	public Action consume(Message me,ConsumeContext cc) {
		if (handler != null && converter != null) {
			try {
				log.info("Get message from sofamq {}", me);
				com.boarsoft.message.bean.Message m = converter.convert(me);
				handler.put(m);
			} catch (Exception e) {
				log.error("Error on handle SofaMQ message", e);
				return Action.ReconsumeLater;
			}
		}
		if (listener != null) {
			listener.consume(me, cc);
		}
		return Action.CommitMessage;
	}

	public MessageService getHandler() {
		return handler;
	}

	public void setHandler(MessageService handler) {
		this.handler = handler;
	}

	public SofaMsgConverter getConverter() {
		return converter;
	}

	public void setConverter(SofaMsgConverter converter) {
		this.converter = converter;
	}
}
